package com.atguigu.gmall.realtime.function;

import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.util.DimUtil;
import com.atguigu.gmall.realtime.util.DruidDSUtil;
import com.atguigu.gmall.realtime.util.RedisUtil;
import com.atguigu.gmall.realtime.util.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import redis.clients.jedis.Jedis;

import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @Author lzc
 * @Date 2022/12/14 09:53
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> {
    
    private ThreadPoolExecutor threadPool;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        // 获取线程池
        threadPool = ThreadPoolUtil.getThreadPool();
    }
    
    public abstract String getTable();
    
    public abstract String getId(T input);
    
    public abstract void addDim(T input, JSONObject dim);
    
    // 参数 1: 输入流中的数据类型
    // 参数 2: 把要输出的数据, 放入到这个参数中
    @Override
    public void asyncInvoke(T input,
                            ResultFuture<T> resultFuture) throws Exception {
        //
        // 数据的所有处理,放入线程池
        threadPool.submit(new Runnable() {
            @Override
            public void run() {
                
                try {
                    Jedis redisClient = RedisUtil.getRedisClient();  // 一个线程获取一个客户端
                    DruidPooledConnection conn = DruidDSUtil.getPhoenixConn();
                    
                    JSONObject dim = DimUtil.readDim(redisClient, conn, getTable(), getId(input));
                    addDim(input, dim);  // 这张表的维度信息补充完毕. 子类去实现
                    
                    // 数据输出到后面的流中
                    resultFuture.complete(Collections.singletonList(input));
                    
                    redisClient.close();
                    conn.close();
                    
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }
        });
        
    }
}
